package com.etc

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * Movie数据集 数据集字段进行分割
  *
  * 1^                           电影ID
  * Toy Story (1995)^            电影的名字
  * In the highlands ....^       电影的描述
  * 85^                          电影的时长
  * August 26, 1997^              电影的发行日期
  * 1995^                         电影的拍摄日期
  * English ^                     电影的语言
  * Action|Drama|Romance|War ^    电影的类型
  * Liam Neeson|Jessica Lange...  电影的演员
  * Michael Caton-Jones           电影的导演
  * tag1|tag2|tag3|....           电影的Tag
  */
case class Movie(val mid: Int, val name: String, val descri: String, val timelong: String, val issue: String,
                 val shoot: String, val language: String, val genres: String, val actors: String, val directives: String)
/**
  * Rating数据集：用户对电影的评分数据集
  * 1                    用户ID
  * 31                   电影ID
  * 2.5                  用户对于电影的评分
  * 1260759144           用户对于电影的评分的时间
  */
case class Rating(val uid:Int,val mid:Int,val score:Double,val timestamp:Int)


/**
  * Mongodb的连接配置
  *
  * @param url
  * @param db
  */
case class MongodbConfig(val url: String, val db: String)

case class Recommendation(rid:Int, r:Double)

/**
  * 电影类别的推荐
  * @param genres   电影的类别
  * @param recs     top10的电影的集合
  */
case class GenresRecommendation(genres:String, recs:Seq[Recommendation])
/**
  * 做分析
  */
object StatisticsRecommender {
  def main(args: Array[String]): Unit = {
    //原数据表
    val MONGODB_MOVIE_COLLECTION = "Movie"
    val MONGODB_RATING_COLLECTION = "Rating"

    //统计的表的名称
    val RATE_MORE_MOVIES = "RateMoreMovies"
    val RATE_MORE_RECENTLY_MOVIES = "RateMoreRecentlyMovies"
    val AVERAGE_MOVIES = "AverageMovies"
    val GENRES_TOP_MOVIES = "GenresTopMovies"

    val config = Map(
      "spark.name" -> "StatisticsRecommender",
      "spark.cores" -> "local[*]",
      "mongodb.uri" -> "mongodb://localhost:27017/movies",
      "mongodb.db" -> "movies"
    )


    //需要创建一个Sparkconf 配置
    val conf = new SparkConf().setMaster(config("spark.cores")).setAppName(config("spark.name"))
    //创建一个sparksession
    val spark = SparkSession.builder().config(conf).getOrCreate()

    import spark.implicits._

    //新建一个mongodb的链接
    val mongodbconfig = MongodbConfig(config("mongodb.uri"), config("mongodb.db"))


    //从mongodb 把评分数据加载至Dataframe
    val ratingDF = spark.read
      //mongodb的地址
      .option("uri", mongodbconfig.url)
      .option("collection", MONGODB_RATING_COLLECTION)
      //取哪个表的数据，在开始（49行）定义了
      //格式化
      .format("com.mongodb.spark.sql")
      .load()
      //下面指的是类型，故用【】,表名
      .as[Rating]
      .toDF

    //从mongodb 把电影数据加载至Dataframe
    val movieDF = spark.read
      //mongodb的地址
      .option("uri", mongodbconfig.url)
      .option("collection", MONGODB_MOVIE_COLLECTION)
      //取哪个表的数据，在开始（48行）定义了
      //格式化
      .format("com.mongodb.spark.sql")
      .load()
      //下面指的是类型，故用【】,表名
      .as[Movie]
      .toDF


    //创建一张名叫ratings的临时视图
    //创建临时视图，此视图的生命周期与用于创建此数据集的[SparkSession]相关联
    ratingDF.createOrReplaceTempView("ratings")

    //创建一张名叫movie的临时视图
    //创建临时视图，此视图的生命周期与用于创建此数据集的[SparkSession]相关联
    movieDF.createOrReplaceTempView("movie")


    //################################### 统计所有历史数据中每个电影的评分 的数量   开始    ##############################//
    // 1  统计所有历史数据中每个电影的评分 的数量
    val rateMoreMoviesDF = spark.sql("select mid,count(mid) from ratings group by mid")
    rateMoreMoviesDF
      .write
      .option("uri",mongodbconfig.url)
      .option("collection",RATE_MORE_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
    //################################### 统计所有历史数据中每个电影的评分 的数量   结束    ##############################//




    //###################################  统计以月为单位每个电影  的评分  的数量   开始    ##############################//
    //2   统计以月为单位每个电影  的评分  的数量

    //创建一个日期格式化工具
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")

    //注册一个UDF函数，用于将timestamp装换成年月格式   1260759144000  => 201605
    spark.udf.register("changeDate",(x:Int) => simpleDateFormat.format(new Date(x * 1000L)).toInt)

    // 将原来的Rating数据集中的时间转换成年月的格式
    val ratingOfYeahMouth = spark.sql("select mid, score, changeDate(timestamp) as yeahmouth from ratings")


    // 将新的数据集注册成为一张表
    ratingOfYeahMouth.createOrReplaceTempView("ratingOfMouth")

    val rateMoreRecentlyMovies = spark.sql("select mid, count(mid) as count ,yeahmouth from ratingOfMouth group by yeahmouth,mid")

    rateMoreRecentlyMovies
        .write
      .option("uri",mongodbconfig.url)
      .option("collection",RATE_MORE_RECENTLY_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    //################################### 统计以月为单位每个电影  的评分  的数量  结束    ##############################//







    //################################### 统计每种电影类型中评分最高的10个电影  开始    ##############################//


    //  3  统计每种电影类型中评分最高的10个电影
    //统计每种电影类型中评分最高的10个电影
    //需要用left join，应为只需要有评分的电影数据集

    //     3.1  统计每个电影的平均评分
    val averageMoviesDF = spark.sql("select mid, avg(score) as avg from ratings group by mid")
    averageMoviesDF
      .write
      .option("uri",mongodbconfig.url)
      .option("collection",AVERAGE_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
    var movieWithScore: DataFrame = movieDF.join(averageMoviesDF, Seq("mid", "mid"))
    //所有的电影类别
    val genres = List("Action","Adventure","Animation","Comedy","Ccrime","Documentary","Drama","Family","Fantasy","Foreign","History","Horror","Music","Mystery"
      ,"Romance","Science","Tv","Thriller","War","Western")

    //将电影类别转换成RDD
    val genresRDD = spark.sparkContext.makeRDD(genres)




    //计算电影类别top10
    val genrenTopMovies = genresRDD.cartesian(movieWithScore.rdd)  //将电影类别和电影数据进行笛卡尔积操作
      .filter{
      // 过滤掉电影的类别不匹配的电影   toLowerCase方法是将大写转为小写
      case (genres,row) => row.getAs[String]("genres").toLowerCase.contains(genres.toLowerCase)
    }
      .map{
        // 将整个数据集的数据量减小，生成RDD[String,Iter[mid,avg]]
        case (genres,row) => {
          //row  取值用  getAs
          (genres,(row.getAs[Int]("mid"), row.getAs[Double]("avg")))
        }
      }.groupByKey()   //将genres数据集中的相同的聚集  得到的是Iterable，所以下面需要toList转为list
      .map{
      // 通过评分的大小进行数据的排序，然后将数据映射为对象
      case (genres, items) => GenresRecommendation(genres,items.toList.sortWith(_._2 > _._2).take(10).map(item => Recommendation(item._1,item._2)))
    }.toDF()

    // 输出数据到MongoDB
    genrenTopMovies
      .write
      .option("uri",mongodbconfig.url)
      .option("collection",GENRES_TOP_MOVIES)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
    //################################### 统计每种电影类型中评分最高的10个电影 结束    ##############################//

    spark.stop()
  }
}
